package sg20.task1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class TaskOneDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //1.get job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "MyDriver");
        job.setJobName("Join University and Country");
        //如果输出文件夹存在,则删除
        FileSystem fs = FileSystem.get(conf);
        if(fs.exists(new Path(args[2])))
            fs.delete(new Path(args[2]), true);
        //2.set jar path
        job.setJarByClass(TaskOneDriver.class);

        //3.connect with class of map and reduce
        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, TaskOneUniversityMapper.class);
        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, TaskOneCountryMapper.class);
        job.setReducerClass(TaskOneReducer.class);

        //4.set the type of K output of map
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //5.set the type of KV output of reduce
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        //6.set input and output path
        FileOutputFormat.setOutputPath(job, new Path(args[2]));


        //7.submit job
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
